#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__

#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <thread>
#include <atomic>

namespace log {
    enum class AsyncType {
        ASYNC_SAFE,// 安全状态，表示缓冲区满了就阻塞，避免资源耗尽的风险
        ASYNC_UNSAFE//不考虑资源耗尽的问题，无限扩容，常用于测试
    };
    class AsyncLooper {
    public:
        using Functor = std::function<void (Buffer &)>;
        using ptr = std::shared_ptr<AsyncLooper>;
    public:
        AsyncLooper(const Functor &cb, AsyncType async_type = AsyncType::ASYNC_SAFE) 
            :stop_flag_(false),
            async_type_(async_type),
            thread_(std::thread(&AsyncLooper::thread_entry, this)),
            callback_(cb) {}
        ~AsyncLooper() {
            stop();
        }
        // 停止工作线程
        void stop() {
            // stop_flag_可能会被同时访问，但直接加锁很浪费，使用atomic修饰保证原子性即可
            stop_flag_ = true;
            con_cond_.notify_all();// 唤醒所有工作线程
            thread_.join();// 等待线程退出
        }
        // 向生产缓冲区写入数据
        void push(const char *data, size_t len) {
            // 1.加锁
            std::unique_lock<std::mutex> lock(mtx_);
            // 2.判断生产缓冲区是否满了，满了就阻塞，不满才写入
            // wait的第二个参数返回结果是true则不阻塞，反之则一直阻塞
            // 考虑缓冲区满了的情况，保证安全，防止无限扩容，用AsyncType::ASYNC_SAFE标识
            if (async_type_ == AsyncType::ASYNC_SAFE)
                pro_cond_.wait(lock, [&](){ return pro_buf_.writeable_size() >= len; });
            // 3.向生产缓冲区写入数据
            pro_buf_.push(data, len);
            // 4.唤醒消费者线程消费
            con_cond_.notify_one();
        }
    private:
        // 线程入口函数
        void thread_entry() {
            while(1) {
                {
                    // 1.获取互斥锁
                    std::unique_lock<std::mutex> lock(mtx_);
                    if (stop_flag_ && pro_buf_.empty()) break;
                    // 2.生产缓冲区有数据或stop_flag_为true时才交换生产缓冲区和消费缓冲区
                    // 否则一直阻塞
                    con_cond_.wait(lock, [&](){ return !pro_buf_.empty() || stop_flag_ == true; });
                    // 3.执行交换
                    con_buf_.swap(pro_buf_);
                    // 4.唤醒所有生产者，可以继续生产
                    pro_cond_.notify_all();
                }
                // 5.调用callback处理消费缓冲区内的数据
                callback_(con_buf_);
                // 6.重置消费缓冲区
                con_buf_.reset();
            }
        }
    private:
        // 工作器停止标志
        // false表示工作（处理数据或阻塞），true表示停止并退出
        std::atomic<bool> stop_flag_;
        AsyncType async_type_;
        // 生产与消费 缓冲区
        Buffer pro_buf_;
        Buffer con_buf_;
        // 互斥锁
        std::mutex mtx_;
        // 生产与消费 条件变量
        std::condition_variable pro_cond_;
        std::condition_variable con_cond_;
        // 回调函数
        Functor callback_;
        // 异步工作器对应的工作线程
        std::thread thread_;
    };
}

#endif 